package com.yuange.promethues.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yuange.promethues.utils.common.HttpUtils;
import com.yuange.promethues.utils.common.TelnetUtil;
import org.springframework.stereotype.Service;

import java.util.HashMap;

@Service
public class Yarn3_1_3ServiceImpl implements Yarn3_1_3Service {

    @Override
    public String getYarnSchedulerMetrics(String addr,Integer port,String impl,String cluster_name) {
        String url = "http://" + addr + ":" + port + impl;

        String yarnMetrics = "";
        JSONObject object = JSONObject.parseObject(HttpUtils.doGet(url, new HashMap<>()));
        if (object != null && object.size() != 0){
            JSONObject queues = object.getJSONObject("scheduler").getJSONObject("schedulerInfo").getJSONObject("queues");
            //获取各队列核心使用百分比
            yarnMetrics = yarnMetrics + "# 各队列核心使用百分比\n";
            JSONArray queueArray = queues.getJSONArray("queue");
            for (Object o : queueArray) {
                JSONObject queue = (JSONObject) o;
                String queueName = queue.getString("queueName");
                JSONObject minEffectiveCapacity = queue.getJSONObject("minEffectiveCapacity");
                JSONObject resourcesUsed = queue.getJSONObject("resourcesUsed");
                Double maxvCores = minEffectiveCapacity.getDouble("vCores");
                Double usedvCores = resourcesUsed.getDouble("vCores");
                String vcores_percentage = "";
                if (maxvCores != 0.0){
                    vcores_percentage = "vcores_percentage{cluster=\""+cluster_name+"\",queue_name=\""+queueName+"\"} " + usedvCores / maxvCores;
                }else {
                    vcores_percentage = "vcores_percentage{cluster=\""+cluster_name+"\",queue_name=\""+queueName+"\"} 0.0";
                }
                yarnMetrics = yarnMetrics + vcores_percentage + "\n";
            }
            //获取各队列内存使用百分比
            yarnMetrics = yarnMetrics + "# 各队列内存使用百分比\n";
            for (Object o : queueArray) {
                JSONObject queue = (JSONObject) o;
                String queueName = queue.getString("queueName");
                JSONObject minEffectiveCapacity = queue.getJSONObject("minEffectiveCapacity");
                JSONObject resourcesUsed = queue.getJSONObject("resourcesUsed");
                Double maxMemory = minEffectiveCapacity.getDouble("memory");
                Double usedMemory = resourcesUsed.getDouble("memory");
                String memory_percentage = "";
                if (maxMemory != 0.0){
                    memory_percentage = "memory_percentage{cluster=\"" + cluster_name+"\",queue_name=\""+queueName+"\"} " + usedMemory / maxMemory;
                }else {
                    memory_percentage = "memory_percentage{cluster=\"" + cluster_name+"\",queue_name=\""+queueName+"\"} 0.0";
                }
                yarnMetrics = yarnMetrics + memory_percentage + "\n";
            }
        }else {
            yarnMetrics = yarnMetrics + "vcores_percentage{cluster=\""+cluster_name+"\",queue_name=\"default\"} 0.0" + "\n";
            yarnMetrics = yarnMetrics + "memory_percentage{cluster=\"" + cluster_name+"\",queue_name=\"default\"} 0.0" + "\n";
        }

        return yarnMetrics;
    }

    @Override
    public String getYarnClusterMetrics(String addr,Integer port,String impl,String cluster_name) {
        String url = "http://" + addr + ":" + port + impl;

        String clusterMetrics = "";
        Integer containersPending = 0;
        Integer appsCompleted = 0;
        Integer containersAllocated = 0;
        Integer rebootedNodes = 0;
        Integer appsSubmitted = 0;
        Integer totalNodes = 0;
        Integer shutdownNodes = 0;
        Integer decommissionedNodes = 0;
        Integer availableMB = 0;
        Integer reservedVirtualCores = 0;
        Integer totalMB = 0;
        Integer allocatedMB = 0;
        Integer decommissioningNodes = 0;
        Integer reservedMB = 0;
        Integer unhealthyNodes = 0;
        Integer appsRunning = 0;
        Integer appsKilled = 0;
        Integer allocatedVirtualCores = 0;
        Integer appsPending = 0;
        Integer activeNodes = 0;
        Integer lostNodes = 0;
        Integer containersReserved = 0;
        Integer totalVirtualCores = 0;
        Integer availableVirtualCores = 0;
        Integer appsFailed = 0;

        JSONObject object = JSONObject.parseObject(HttpUtils.doGet(url, new HashMap<>()));
        if (object != null && object.size() != 0){
            JSONObject clusterMetricsObj = object.getJSONObject("clusterMetrics");
            //获取cluster集群指标
            clusterMetrics = clusterMetrics + "# cluster集群指标\n";

            //仍在等待分配的所有容器的总数
            clusterMetrics = clusterMetrics + "# 仍在等待分配的所有容器的总数\n";
            containersPending = clusterMetricsObj.getInteger("containersPending");

            //已提交，已完成运行或失败的应用程序总数
            //已完成运行的应用程序总数
            clusterMetrics = clusterMetrics + "# 已完成运行的应用程序总数\n";
            appsCompleted = clusterMetricsObj.getInteger("appsCompleted");

            //已分配的所有容器的总数
            clusterMetrics = clusterMetrics + "# 已分配的所有容器的总数\n";
            containersAllocated = clusterMetricsObj.getInteger("containersAllocated");

            //系统中丢失，重新启动或运行状况不佳的所有节点的总数
            //系统中重新启动的所有节点的总数
            clusterMetrics = clusterMetrics + "# 系统中重新启动的所有节点的总数\n";
            rebootedNodes = clusterMetricsObj.getInteger("rebootedNodes");

            //已提交的应用程序总数
            clusterMetrics = clusterMetrics + "# 已提交的应用程序总数\n";
            appsSubmitted = clusterMetricsObj.getInteger("appsSubmitted");

            //所有节点的总数
            clusterMetrics = clusterMetrics + "# 所有节点的总数\n";
            totalNodes = clusterMetricsObj.getInteger("totalNodes");

            //所有节点中已停止运行的数量
            clusterMetrics = clusterMetrics + "# 所有节点中已停止的运行的数量\n";
            shutdownNodes = clusterMetricsObj.getInteger("shutdownNodes");

            //系统中已停用的节点数量
            clusterMetrics = clusterMetrics + "# 系统中已停用的节点数量\n";
            decommissionedNodes = clusterMetricsObj.getInteger("decommissionedNodes");

            //可利用的内存总量
            clusterMetrics = clusterMetrics + "# 可利用的内存总量\n";
            availableMB = clusterMetricsObj.getInteger("availableMB");

            //申请中的核心数
            clusterMetrics = clusterMetrics + "# 申请中的核心数\n";
            reservedVirtualCores = clusterMetricsObj.getInteger("reservedVirtualCores");

            //总内存数
            clusterMetrics = clusterMetrics + "# 总内存数\n";
            totalMB = clusterMetricsObj.getInteger("totalMB");

            //已使用的内存数
            clusterMetrics = clusterMetrics + "# 已使用的内存数\n";
            allocatedMB = clusterMetricsObj.getInteger("allocatedMB");

            //停用的节点数
            clusterMetrics = clusterMetrics + "# 停用的节点数\n";
            decommissioningNodes = clusterMetricsObj.getInteger("decommissioningNodes");

            //申请中的内存数
            clusterMetrics = clusterMetrics + "# 申请中的内存数\n";
            reservedMB = clusterMetricsObj.getInteger("reservedMB");

            //状态不良的节点数
            clusterMetrics = clusterMetrics + "# 状态不良的节点数\n";
            unhealthyNodes = clusterMetricsObj.getInteger("unhealthyNodes");

            //正在运行的应用程序数
            clusterMetrics = clusterMetrics + "# 正在运行的应用程序数\n";
            appsRunning = clusterMetricsObj.getInteger("appsRunning");

            //已杀死的应用程序数
            clusterMetrics = clusterMetrics + "# 已杀死的应用程序数\n";
            appsKilled = clusterMetricsObj.getInteger("appsKilled");

            //已使用的核心数
            clusterMetrics = clusterMetrics + "# 已使用的核心数\n";
            allocatedVirtualCores = clusterMetricsObj.getInteger("allocatedVirtualCores");

            //待处理的应用程序数
            clusterMetrics = clusterMetrics + "# 待处理的应用程序数\n";
            appsPending = clusterMetricsObj.getInteger("appsPending");

            //活动中的节点数
            clusterMetrics = clusterMetrics + "# 活动中的节点数\n";
            activeNodes = clusterMetricsObj.getInteger("activeNodes");

            //丢失的节点数
            clusterMetrics = clusterMetrics + "# 丢失的节点数\n";
            lostNodes = clusterMetricsObj.getInteger("lostNodes");

            //保留的容器总数
            clusterMetrics = clusterMetrics + "# 保留的容器总数\n";
            containersReserved = clusterMetricsObj.getInteger("containersReserved");

            //总虚拟核数
            clusterMetrics = clusterMetrics + "# 总虚拟核数\n";
            totalVirtualCores = clusterMetricsObj.getInteger("totalVirtualCores");

            //可利用的虚拟核数
            clusterMetrics = clusterMetrics + "# 可利用的虚拟核数\n";
            availableVirtualCores = clusterMetricsObj.getInteger("availableVirtualCores");

            //失败的应用程序数
            clusterMetrics = clusterMetrics + "# 失败的应用程序数\n";
            appsFailed = clusterMetricsObj.getInteger("appsFailed");
        }

        clusterMetrics = clusterMetrics + "yarn_containerspending{cluster=\"" + cluster_name + "\"} "+ containersPending +"\n";
        clusterMetrics = clusterMetrics + "yarn_appscompleted{cluster=\"" + cluster_name + "\"} "+ appsCompleted +"\n";
        clusterMetrics = clusterMetrics + "yarn_acontainersallocated{cluster=\"" + cluster_name + "\"} "+ containersAllocated +"\n";
        clusterMetrics = clusterMetrics + "yarn_rebootedNodes{cluster=\"" + cluster_name + "\"} "+ rebootedNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_appssubmitted{cluster=\"" + cluster_name + "\"} "+ appsSubmitted +"\n";
        clusterMetrics = clusterMetrics + "yarn_totalnodes{cluster=\"" + cluster_name+ "\"} "+ totalNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_shutdownnodes{cluster=\"" + cluster_name + "\"} "+ shutdownNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_decommissionednodes{cluster=\"" + cluster_name + "\"} "+ decommissionedNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_availablemb{cluster=\"" + cluster_name + "\"} "+ availableMB +"\n";
        clusterMetrics = clusterMetrics + "yarn_reservedvirtualcores{cluster=\"" + cluster_name + "\"} "+ reservedVirtualCores +"\n";
        clusterMetrics = clusterMetrics + "yarn_totalmb{cluster=\"" + cluster_name + "\"} "+ totalMB +"\n";
        clusterMetrics = clusterMetrics + "yarn_allocatedmb{cluster=\"" + cluster_name + "\"} "+ allocatedMB +"\n";
        clusterMetrics = clusterMetrics + "yarn_decommissioningnodes{cluster=\"" + cluster_name + "\"} "+ decommissioningNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_reservedmb{cluster=\"" + cluster_name + "\"} "+ reservedMB +"\n";
        clusterMetrics = clusterMetrics + "yarn_unhealthynodes{cluster=\"" + cluster_name+ "\"} "+ unhealthyNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_appsRunning{cluster=\"" + cluster_name + "\"} "+ appsRunning +"\n";
        clusterMetrics = clusterMetrics + "yarn_appskilled{cluster=\"" + cluster_name+ "\"} "+ appsKilled +"\n";
        clusterMetrics = clusterMetrics + "yarn_allocatedvirtualcores{cluster=\"" + cluster_name + "\"} "+ allocatedVirtualCores +"\n";
        clusterMetrics = clusterMetrics + "yarn_appspending{cluster=\"" + cluster_name + "\"} "+ appsPending +"\n";
        clusterMetrics = clusterMetrics + "yarn_activenodes{cluster=\"" + cluster_name + "\"} "+ activeNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_lostnodes{cluster=\"" + cluster_name + "\"} "+ lostNodes +"\n";
        clusterMetrics = clusterMetrics + "yarn_containersreserved{cluster=\"" + cluster_name + "\"} "+ containersReserved +"\n";
        clusterMetrics = clusterMetrics + "yarn_totalvirtualcores{cluster=\"" + cluster_name + "\"} "+ totalVirtualCores +"\n";
        clusterMetrics = clusterMetrics + "yarn_availablevirtualcores{cluster=\"" + cluster_name + "\"} "+ availableVirtualCores +"\n";
        clusterMetrics = clusterMetrics + "yarn_appsfailed{cluster=\"" + cluster_name + "\"} "+ appsFailed +"\n";

        return clusterMetrics;
    }

    @Override
    public String getNameNodeJMX(String addr,Integer port,String impl,String type_name) {
        String url = "http://" + addr + ":" + port + impl;

        String nameNodeJMX = "";

        JSONObject object = JSONObject.parseObject(HttpUtils.doGet(url, new HashMap<>()));
        if (object != null && object.size() != 0){
            JSONArray beans = object.getJSONArray("beans");
            //NameNode中的NumStaleDataNodes指标
            nameNodeJMX = nameNodeJMX + "# NameNode中的NumStaleDataNodes指标\n";
            for (Object o : beans) {
                JSONObject bean = (JSONObject) o;
                Double numStaleDataNodes = bean.getDouble("NumStaleDataNodes");
                if (numStaleDataNodes != null){
                    nameNodeJMX = nameNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"NumStaleDataNodes\",port=\""
                            +port+"\",type=\""+type_name+"\"} "
                            +numStaleDataNodes+"\n";
                    break;
                }
            }
            //NameNode中的HeapMemoryUsage指标、NonHeapMemoryUsage指标
            for (Object o : beans) {
                JSONObject bean = (JSONObject) o;
                JSONObject heapMemoryUsage = bean.getJSONObject("HeapMemoryUsage");
                JSONObject nonHeapMemoryUsage = bean.getJSONObject("NonHeapMemoryUsage");
                if (heapMemoryUsage != null && heapMemoryUsage.size() != 0
                        && nonHeapMemoryUsage != null && nonHeapMemoryUsage.size() != 0){
                    Double usedHeapMemoryUsage = heapMemoryUsage.getDouble("used");
                    Double usedNonHeapMemoryUsage = nonHeapMemoryUsage.getDouble("used");
                    nameNodeJMX = nameNodeJMX + "# NameNode中的HeapMemoryUsage指标\n";
                    nameNodeJMX = nameNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"HeapMemoryUsage\",port=\""
                            +port+"\",type=\""+type_name+"\"} "
                            +(usedHeapMemoryUsage/ 1024 / 1024)+"\n";
                    nameNodeJMX = nameNodeJMX + "# NameNode中的NonHeapMemoryUsage指标\n";
                    nameNodeJMX = nameNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"NonHeapMemoryUsage\",port=\""
                            +port+"\",type=\""+type_name+"\"} "
                            +(usedNonHeapMemoryUsage/ 1024 / 1024)+"\n";
                    break;
                }
            }
        }else {
            nameNodeJMX = nameNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"NumStaleDataNodes\",port=\""
                    +port+"\",type=\""+type_name+"\"} "
                    +0.0+"\n";
            nameNodeJMX = nameNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"HeapMemoryUsage\",port=\""
                    +port+"\",type=\""+type_name+"\"} "
                    +(0.0)+"\n";
            nameNodeJMX = nameNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"NonHeapMemoryUsage\",port=\""
                    +port+"\",type=\""+type_name+"\"} "
                    +(0.0)+"\n";
        }

        return nameNodeJMX;
    }

    @Override
    public String getHiveServer2JMX(String addr,Integer port,String impl,String type_name) {
        String url = "http://" + addr + ":" + port + impl;

        String hiveServer2JMX = "";

        String s = HttpUtils.doGet(url, new HashMap<>());
        if (s != null && s.length() != 0){
            JSONObject object = JSONObject.parseObject(s.replaceAll("-Infinity","0.0"));
            JSONArray beans = object.getJSONArray("beans");
            //hiveServer2中的HeapMemoryUsage指标、NonHeapMemoryUsage指标
            for (Object o : beans) {
                JSONObject bean = (JSONObject) o;
                JSONObject heapMemoryUsage = bean.getJSONObject("HeapMemoryUsage");
                JSONObject nonHeapMemoryUsage = bean.getJSONObject("NonHeapMemoryUsage");
                if (heapMemoryUsage != null && heapMemoryUsage.size() != 0
                        && nonHeapMemoryUsage != null && nonHeapMemoryUsage.size() != 0){
                    Double usedHeapMemoryUsage = heapMemoryUsage.getDouble("used");
                    Double usedNonHeapMemoryUsage = nonHeapMemoryUsage.getDouble("used");
                    hiveServer2JMX = hiveServer2JMX + "# hiveServer2中的HeapMemoryUsage指标\n";
                    hiveServer2JMX = hiveServer2JMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"HeapMemoryUsage\",port=\""
                            +port+"\",type=\""+type_name+"\"} "
                            +(usedHeapMemoryUsage/ 1024 / 1024)+"\n";
                    hiveServer2JMX = hiveServer2JMX + "# hiveServer2中的NonHeapMemoryUsage指标\n";
                    hiveServer2JMX = hiveServer2JMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"NonHeapMemoryUsage\",port=\""
                            +port+"\",type=\""+type_name+"\"} "
                            +(usedNonHeapMemoryUsage/ 1024 / 1024)+"\n";
                    break;
                }
            }
        }else {
            hiveServer2JMX = hiveServer2JMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"HeapMemoryUsage\",port=\""
                    +port+"\",type=\""+type_name+"\"} "
                    +(0.0)+"\n";
            hiveServer2JMX = hiveServer2JMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"NonHeapMemoryUsage\",port=\""
                    +port+"\",type=\""+type_name+"\"} "
                    +(0.0)+"\n";
        }

        return hiveServer2JMX;
    }

    @Override
    public String getDataNodeJMX(String addr, Integer port, String impl, String type_name) {
        String url = "http://" + addr + ":" + port + impl;

        String dataNodeJMX = "";

        JSONObject object = JSONObject.parseObject(HttpUtils.doGet(url, new HashMap<>()));
        if (object != null && object.size() != 0){
            JSONArray beans = object.getJSONArray("beans");

            //DataNode中的HeartbeatsTotalNumOps指标
            dataNodeJMX = dataNodeJMX + "# DataNode中的HeartbeatsTotalNumOps指标\n";
            for (Object o : beans) {
                JSONObject bean = (JSONObject) o;
                Double heartbeatsTotalNumOps = bean.getDouble("HeartbeatsTotalNumOps");
                if (heartbeatsTotalNumOps != null){
                    dataNodeJMX = dataNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"HeartbeatsTotalNumOps\",port=\""
                            +port+"\",type=\""+type_name+"\"} "
                            +heartbeatsTotalNumOps+"\n";
                    break;
                }
            }
        }else {
            dataNodeJMX = dataNodeJMX + "ACTIVESTATE{ip=\""+addr+"\",name=\"HeartbeatsTotalNumOps\",port=\""
                    +port+"\",type=\""+type_name+"\"} 0.0\n";
        }
        return dataNodeJMX;
    }

    @Override
    public String getTelnetJMX(String ip, Integer port, String component_name, String type_name) {
        String telnetJMX = "# telnet方式判断IP端口是否存活，存活值为1.0，死亡值为0.0\n";
        if (TelnetUtil.telnet(ip, port)){
            telnetJMX = telnetJMX + "ACTIVESTATE{ip=\""+ip+"\",name=\""+component_name+"\",port=\""+port
                    +"\",type=\""+type_name+"\"} 1.0\n";
        }else {
            telnetJMX = telnetJMX + "ACTIVESTATE{ip=\""+ip+"\",name=\""+component_name+"\",port=\""+port
                    +"\",type=\""+type_name+"\"} 0.0\n";
        }
        return telnetJMX;
    }
}
